package streaming.day01

import java.sql.DriverManager

import com.typesafe.config.ConfigFactory
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * sheep.Old @ 64341393
  * Created 2018/4/21
  */
object WordCountStatePlus {

    def main(args: Array[String]): Unit = {

        // 创建一个解析配置文件的对象
        val config = ConfigFactory.load()

        // master 需要 2 个核, 以防止饥饿情况（starvation scenario）.
        // 1一个核用于运行当前程序，一个核用于接收数据
        val conf = new SparkConf()
          .setAppName("wordcountshow")
          .setMaster("local[*]")

        // 第二个参数为批次时间间隔，多长时间的数据集为一个批次，这个时间不能拍脑门设置，科学合理的设置,一般会比较短
        val ssc = new StreamingContext(conf, Seconds(2)) // 2秒钟一个批次


        // 接收受数据
        val stream: ReceiverInputDStream[String] = ssc.socketTextStream("10.172.50.12", 44444)


        stream.foreachRDD(rdd => {


            // 计算当前批次结果
            val current_batch_result: RDD[(String, Int)] = rdd.flatMap(_.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)

            // 插入当前批次计算出来的数据结果
            current_batch_result.foreachPartition(partition => {
                // 创建一个连接
                val url = config.getString("db.url")
                val user = config.getString("db.user")
                val passwd = config.getString("db.password")
                val conn = DriverManager.getConnection(url, user, passwd)

                // 将当前分区里面的所有数据都插入到数据库中
                partition.foreach(tp => {
                    // 判断即将要插入的数据是否之前已经插入过了，如果已经插入过了，则本次应该是更新动作，否则就是插入
                    val pstmts = conn.prepareStatement("select * from wordcount where words=?")
                    pstmts.setString(1, tp._1)
                    val rs = pstmts.executeQuery()

                    var flag = false
                    while (rs.next()) {

                        flag = true // 已存在单词

                        // 即将要插入的单词已经存在了，可以断定我们要开始更新操作了
                        val dbCurrentCount = rs.getInt("total")
                        // 计算最新的值
                        val newCount = dbCurrentCount + tp._2

                        // 更新了
                        val update = conn.prepareStatement("update wordcount set total=? where words=?")
                        update.setInt(1, newCount)
                        update.setString(2, tp._1)

                        update.executeUpdate()
                        update.close()
                    }
                    rs.close()
                    pstmts.close()


                    if(!flag){
                        // 插入每一条数据
                        val pstmti = conn.prepareStatement("insert into wordcount values(?,?)")
                        pstmti.setString(1, tp._1)
                        pstmti.setInt(2, tp._2)
                        // 执行sql
                        pstmti.executeUpdate()
                        pstmti.close()
                    }
                })

                if(null != conn) conn.close()
            })
        })


        // 启动程序
        ssc.start() // 启动当前main方法这个主程序
        // 不要死
        ssc.awaitTermination() // 主程序阻塞
    }

}
